package cn.ymatrix.mxgateSDKTOYMatrix;

import cn.ymatrix.apiclient.DataPostListener;
import cn.ymatrix.apiclient.MxClient;
import cn.ymatrix.apiclient.Result;
import cn.ymatrix.builder.MxBuilder;
import cn.ymatrix.builder.RequestType;
import cn.ymatrix.data.Tuple;
import cn.ymatrix.logger.LoggerLevel;
import cn.ymatrix.logger.MxLogger;
import cn.ymatrix.utils.Config;
import cn.ymatrix.utils.DataUtil;
import org.apache.commons.lang3.time.StopWatch;
import org.slf4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class MxgateSDKTOYMatrix {

  private static final String CUSTOMER_LOG_TAG = "[>>>CUSTOMER<<<] ";
  private static MxBuilder mxBuilder;

  public static void main(String[] args) {

    MxLogger.loggerLevel(LoggerLevel.INFO);
    Logger LOGGER = MxLogger.init(MxgateSDKTOYMatrix.class);

    // 初始化Mxgate-SDK的配置
    initMxgateSDKBuilder(LOGGER);

    // 加载数据到YMatrix中
    syncMxgateDataLoad(LOGGER);
  }

  public static void syncMxgateDataLoad(Logger LOGGER) {
    String requestType = Config.getRequestType();
    String httpHost = Config.getHttpUrl();
    String gRPCHost = Config.getgRPCHost();
    String schema = Config.getSchemaName();
    String table = Config.getTableName();
    try {
      if (requestType.equals("http")) {
        MxClient client = mxBuilder.connect(httpHost, gRPCHost, schema, table);
        sendDataToMxgate(LOGGER, client);
      } else if (requestType.equals("grpc")) {
        MxClient client = mxBuilder.connect(gRPCHost, gRPCHost, schema, table);
        sendDataToMxgate(LOGGER, client);
      }

    } catch (Exception e) {
      LOGGER.error("MxClient init error: {}" + e.getMessage());
      e.printStackTrace();
    }
  }

  private static void sendDataToMxgate(Logger LOGGER, MxClient client) {
    // private static void sendDataToMxgate(Logger LOGGER, MxClient client) {
    String linesOrBytesToFlush = Config.getLinesOrBytesToFlush();
    int totalNumber = Config.getTotalNumber();
    int batchSize = Config.getBatchSize();
    int enoughLinesOrBytes = Config.getEnoughLinesOrBytes();
    int csvConstructionBatchSize = Config.getCsvConstructionBatchSize();
    String compressWithZstd = Config.getCompressWithZstd();
    String requestType = Config.getRequestType();
    StopWatch startWatch = StopWatch.createStarted();

    if (client != null) {
      // client.withoutCompress();
      if (csvConstructionBatchSize > 0) {
        client.withCSVConstructionBatchSize(csvConstructionBatchSize);
      }
      if (linesOrBytesToFlush.equals("lines")) {
        // NOTE: 使用按行 flush 的速度要比按 bytes 快。
        client.withEnoughLinesToFlush(enoughLinesOrBytes);
      } else if (linesOrBytesToFlush.equals("bytes")) {
        client.withEnoughBytesToFlush(enoughLinesOrBytes);
      } else {
        LOGGER.error("Invalid flush mode :" + linesOrBytesToFlush);
        return;
      }
      if (compressWithZstd.equals("zstd")) {
        LOGGER.info("with compress");
        client.withCompress();
        if (requestType.equals("grpc")) {
          LOGGER.info("with base64");
          client.withBase64Encode4Compress();
        }
      }
      client.registerDataPostListener(
          new DataPostListener() {
            public void onSuccess(Result result) {}

            public void onFailure(Result result) {
              LOGGER.error(
                  CUSTOMER_LOG_TAG
                      + "Send tuples fail msg: {} lines: {}"
                      + result.getMsg()
                      + result.getSucceedLines());
              LOGGER.error(
                  CUSTOMER_LOG_TAG
                      + "Sendtuples fail error tuples:{} "
                      + result.getErrorTuplesMap());
              for (Map.Entry<Tuple, String> entry : result.getErrorTuplesMap().entrySet()) {
                LOGGER.error(
                    CUSTOMER_LOG_TAG
                        + "error tuple of table="
                        + entry.getKey().getTableName()
                        + "tuple="
                        + entry.getKey()
                        + "reason="
                        + entry.getValue());
              }
            }
          });

      int insertCount = 0;
      int counter = 0;
      LOGGER.info("---- 开始插入数据 ----");
      for (counter = 1; counter <= totalNumber; counter++) {
        if (counter % batchSize == 0) {
          insertCount++;
          sendDataTOTuple(client, batchSize);
          LOGGER.info("成功插入第 " + counter + " 条数据");
        }
      }
      int tailCount = totalNumber - insertCount * batchSize;
      if (tailCount != 0) {
        sendDataTOTuple(client, tailCount);
      }
      LOGGER.info(
          "成功插入 " + totalNumber + " 条数据,耗时: " + startWatch.getTime(TimeUnit.MILLISECONDS) + " 毫秒");

      try {
        Thread.sleep(5000L);
        System.exit(0);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }

  private static void initMxgateSDKBuilder(Logger LOGGER) {
    String requestType = Config.getRequestType();
    String dropAll = Config.getDropAll();
    String asyncMode = Config.getAsyncMode();

    MxBuilder.Builder builder =
        MxBuilder.newBuilder()
            .withDropAll(false)
            .withCacheCapacity(Config.getCacheCapacity())
            .withCacheEnqueueTimeout(Config.getCacheEnqueueTimeout())
            .withConcurrency(Config.getSdkConcurrency())
            .withRequestTimeoutMillis(Config.getRequestTimeoutMillis())
            .withMaxRequestQueued(Config.getMaxRequestQueued())
            .withMaxRetryAttempts(Config.getMaxRetryAttempts())
            .withRetryWaitDurationMillis(Config.getRetryWaitDurationMillis());

    if (requestType.equals("http")) {
      builder.withRequestType(RequestType.WithHTTP);
    } else if (requestType.equals("grpc")) {
      builder.withRequestType(RequestType.WithGRPC);
    } else {
      LOGGER.error("Invalid request type, should be http or grpc");
      return;
    }
    if (dropAll.equals("dropAll")) {
      builder.withDropAll(true);
    }
    if (asyncMode.equals("async")) {
      builder.withRequestAsync(true);
    }
    mxBuilder = builder.build();
  }

  private static void sendDataTOTuple(MxClient client, int batchSize) {
    List<Tuple> listTuple = new ArrayList<>();
    for (int i = 0; i < batchSize; i++) {
      Tuple tuple = client.generateEmptyTupleLite();
      tuple.addColumn("ts", DataUtil.getCurrentTimestamp());
      tuple.addColumn("vin", DataUtil.getIntValue());
      tuple.addColumn("t1", DataUtil.getIntValue());
      tuple.addColumn("t2", DataUtil.getIntValue());
      tuple.addColumn("t3", DataUtil.getIntValue());
      tuple.addColumn("t4", DataUtil.getIntValue());
      listTuple.add(tuple);
    }
    client.appendTuplesList(listTuple);
  }
}
